use crate::cache::IndexChange;
use crate::keys::VddKey;
use crate::record_rule_db::RecordRuleDbTrait;
use crate::subscribe::{self, FiledValueChange};
use crate::value_db::{UniqueChange, VddValueDdTrait};
use crate::{
    cache, delete_by_key_head, model_db, move_to_trash, record_rule_db, record_version_db, CF_FV,
    CF_TRASH, ROCKSDB,
};
use crate::{log_db, visit_db};
use crate::{value_db, CF_META};
use rust_rocksdb::{TransactionDB, WriteBatchWithTransaction};
use serde_json::json;
use std::collections::HashMap;
use std::sync::Arc;
use visit_dd_core::field::VddFieldType;
use visit_dd_core::i18n_text::E_ROCKSDB;
use visit_dd_core::model::{VddModel, VddSchema};
use visit_dd_core::proj::VddLogPart;
use visit_dd_core::query::{VddValueEvaluation, WebQueryParams};
use visit_dd_core::record::{
    RecordConsume, RecordMeta, RecordRule, RecordRuleDetail, VddDisplay, VddRecord,
    VddRecordInTrash,
};
use visit_dd_core::value::{VddValue, VddValueRef, VddValueWithMeta};
use visit_dd_core::visit::VisitGroup;
use visit_dd_core::{
    alog, bool_to_ares, extract_u64_from_key, i18n, to_sha256_hash, Ares, Astr, FieldId, RecordId,
    UserId, AC_ADD, AC_DELETE, AC_EDIT, EMPTY_RID,
};

pub(crate) trait UniKeyTraitExtern {
    fn get_unique_key_by_fids(&self, rvr: &RecordRule) -> String;
    fn delete_unique_after_delete_record(
        &self,
        v_rules: &[Arc<RecordRule>],
        changes: &mut Vec<UniqueChange>,
    );
}

impl UniKeyTraitExtern for VddRecord {
    fn get_unique_key_by_fids(&self, rvr: &RecordRule) -> String {
        let mut raw_value = String::new();
        let fid_s = rvr.get_fids();
        fid_s.iter().for_each(|x| {
            raw_value.push_str("_");
            raw_value.push_str(
                &self
                    .value
                    .get(x)
                    .unwrap_or(&Arc::new(VddValue::NULL))
                    .get_simple_display(),
            )
        });
        let hash = to_sha256_hash(raw_value.as_bytes());
        VddKey::UniqueRecordValue(rvr.id, hash).to_string()
    }
    /// 删除记录中的唯一字段的hash值占用
    fn delete_unique_after_delete_record(
        &self,
        v_rules: &[Arc<RecordRule>],
        changes: &mut Vec<UniqueChange>,
    ) {
        for v_rule in v_rules {
            match v_rule.detail.as_ref() {
                RecordRuleDetail::RrUnique(..) => {
                    let unique_key = self.get_unique_key_by_fids(&v_rule);
                    changes.push(UniqueChange {
                        a: u8::MAX,
                        key: unique_key.into(),
                        value: self.meta.id.0,
                    });
                }
                _ => {}
            }
        }
        for field_id in self.value.keys() {
            // unique limit of field
            let unique_key = self.get_value(&field_id).get_unique_key(&field_id);
            changes.push(UniqueChange {
                a: u8::MAX,
                key: unique_key.into(),
                value: self.meta.id.0,
            });
        }
    }
}

pub trait RecordDbTrait {
    fn fit_record_rules(
        &self,
        schema: &VddSchema,
        v_rules: &[Arc<RecordRule>],
        is_update_record: bool,
    ) -> impl std::future::Future<Output = Ares> + Send;
    fn fit_field_defines(&mut self, schema: &VddSchema, is_update_record: bool) -> Ares;
    fn new_display(
        &self,
        schema: &VddSchema,
    ) -> impl std::future::Future<Output = VddDisplay> + Send;
    fn write_rdb(
        &self,
        schema: &VddSchema,
        batch: &mut WriteBatchWithTransaction<true>,
        index_changes: &mut Vec<IndexChange>,
    );
    fn save_unique_key(
        &self,
        schema: &VddSchema,
        v_rules: &[Arc<RecordRule>],
        changes: &mut Vec<UniqueChange>,
    );
}

impl RecordDbTrait for VddRecord {
    /// 校验唯一字段是否满足唯一限制
    async fn fit_record_rules(
        &self,
        schema: &VddSchema,
        v_rules: &[Arc<RecordRule>],
        is_update_record: bool,
    ) -> Ares {
        for record_rule in v_rules {
            // println!("{}", record_rule.get_display().await);
            let r = match record_rule.detail.as_ref() {
                RecordRuleDetail::RrUnique(..) => {
                    let unique_key = self.get_unique_key_by_fids(record_rule);
                    let uni_v = value_db::unique_check(&unique_key);
                    let b = uni_v == 0 || (is_update_record && uni_v == self.meta.id.0);
                    bool_to_ares(b)
                }
                RecordRuleDetail::RrFitEvaluation(fve) => bool_to_ares(self.fit_evaluation(fve)),
                RecordRuleDetail::RrConsumeTimes { consumer_m_id, max } => {
                    if *consumer_m_id == self.meta.model_id {
                        let c_field = schema.get_field_by_provider(record_rule.model_id, None);
                        if c_field.is_none() {
                            Ok(())
                        } else {
                            let c_field_id = c_field.unwrap().id.clone();
                            // B模型中的记录ID
                            let p_rec_id: RecordId =
                                self.get_value(&c_field_id).as_ref().try_into().unwrap();
                            // 查B模型中的记录在本模型内的消费情况
                            let consumes = record_rule_db::rc_select(
                                p_rec_id,
                                self.meta.model_id,
                                Some(c_field_id),
                            );
                            if *max > 0 && consumes.len() >= *max {
                                Err(format!("已达到最多消费次数 {}", *max).into())
                            } else {
                                Ok(())
                            }
                        }
                    } else {
                        Ok(())
                    }
                }
            };
            if r.is_err() {
                return Err(
                    format!("记录不满足如下规则 {}", record_rule.get_display().await).into(),
                );
            }
        }
        Ok(())
    }

    /// 校验记录是否符合字段定义的要求 同时填充自动加减值计算字段
    fn fit_field_defines(&mut self, schema: &VddSchema, is_update_record: bool) -> Ares {
        // tracing::info!("{:?}",fields);
        for field in schema.fields.iter() {
            // in loop, only can return error, do not return ok!
            // tracing::info!("{:?}", field);
            let v = self.get_value(&field.id);
            if field.required && !field.auto_generated() && v.as_ref() == &VddValue::NULL {
                return Err(i18n!(1122, &field.label));
            }
            if field.unique && !field.auto_generated() {
                // 唯一性校验
                let unique_key = v.get_unique_key(&field.id);
                let r = value_db::unique_check(&unique_key);
                if r > 0 {
                    if !is_update_record || (is_update_record && self.meta.id.0 != r) {
                        return Err(i18n!(1132, &field.label));
                    }
                }
            }
            match field.field_type.as_ref() {
                VddFieldType::FtString { .. } => match v.as_ref() {
                    VddValue::S101(v2) => field.try_fit_str(v2)?,
                    VddValue::M101(v2) => {
                        for item in v2 {
                            field.try_fit_str(item)?
                        }
                    }
                    VddValue::NULL => {}
                    _ => return Err(i18n!(1133)),
                },
                VddFieldType::FtJson { .. } => match v.as_ref() {
                    VddValue::S102(v2) => field.try_fit_str(v2)?,
                    VddValue::NULL => {}
                    _ => return Err(i18n!(1133)),
                },
                VddFieldType::FtInt { .. } => match v.as_ref() {
                    VddValue::S201(v2) => field.fit_201(*v2)?,
                    VddValue::M201(v2) => {
                        for item in v2 {
                            field.fit_201(*item)?
                        }
                    }
                    VddValue::NULL => {}
                    _ => return Err(i18n!(1133)),
                },
                VddFieldType::FtFloat { .. } => match v.as_ref() {
                    VddValue::S202(v2) => field.fit_202(*v2)?,
                    VddValue::M202(v2) => {
                        for item in v2 {
                            field.fit_202(*item)?
                        }
                    }
                    VddValue::NULL => {}
                    _ => return Err(i18n!(1133)),
                },
                VddFieldType::FtAutoIncreaseInt { start, .. } => {
                    let need_gen = if let Ok(v212) =
                        TryInto::<i64>::try_into(self.value.get(&field.id).unwrap().as_ref())
                    {
                        v212 == *start
                    } else {
                        true
                    };
                    if need_gen {
                        let fv = VddValue::S201(cache::fetch_auto_increment_id(&field));
                        self.value.insert(field.id.clone(), fv.into());
                    }
                }
                VddFieldType::FtAutoIncreaseIntGlobal => {
                    let need_gen = if let Ok(v) =
                        TryInto::<u64>::try_into(self.value.get(&field.id).unwrap().as_ref())
                    {
                        v < 1
                    } else {
                        true
                    };
                    if need_gen {
                        let fv = VddValue::S291(cache::fetch_system_auto_increment_id());
                        self.value.insert(field.id.clone(), fv.into());
                    }
                }
                _ => {}
            }
        }
        Ok(())
    }

    /// record_display_db是私有包，所以在此包中包装一层调用
    async fn new_display(&self, schema: &VddSchema) -> VddDisplay {
        let mut recordd = VddDisplay {
            meta: self.meta.clone(),
            value: self.value.clone(),
            display: HashMap::<FieldId, Astr>::new(),
        };
        display::fill_display(&mut recordd, schema, true).await;
        recordd
    }

    /// 通用的写入记录的方法
    fn write_rdb(
        &self,
        schema: &VddSchema,
        batch: &mut WriteBatchWithTransaction<true>,
        index_changes: &mut Vec<IndexChange>,
    ) {
        // 保存基础信息
        let cf_meta = ROCKSDB.cf_handle(CF_META).unwrap();
        let obj_key = VddKey::EntityRecordMeta(self.meta.id.clone()).to_string();
        batch.put_cf(&cf_meta, obj_key, json!(&self.meta).to_string());
        // 保存各项值
        self.value.iter().for_each(|x| {
            let cf_fv = ROCKSDB.cf_handle(CF_FV).unwrap();
            batch.put_cf(
                &cf_fv,
                VddKey::BoundFieldValue(x.0.clone(), self.meta.id.clone()).to_string(),
                json!(x.1).to_string(),
            );
            // 保存消费关系
            match x.1.as_ref() {
                VddValue::S901(p_rec_id) => {
                    schema
                        .fields
                        .iter()
                        .find(|field| &field.id == x.0)
                        .map(|field| {
                            record_rule_db::rc_save(
                                batch,
                                &RecordConsume {
                                    record_id: p_rec_id.clone(),
                                    consumer_m_id: self.meta.model_id,
                                    consumer_f_id: field.id.clone(),
                                    consumer_r_id: self.meta.id.clone(),
                                },
                            );
                        });
                }
                _ => {}
            }
        });
        index_changes.push(IndexChange::new(
            VddKey::IndexModel2Record(self.meta.model_id),
            AC_ADD,
            self.meta.id.0,
        ));
    }

    /// 记录保存后完善唯一键
    fn save_unique_key(
        &self,
        schema: &VddSchema,
        v_rules: &[Arc<RecordRule>],
        changes: &mut Vec<UniqueChange>,
    ) {
        for field in schema.fields.iter() {
            if field.unique {
                let v = self.get_value(&field.id);
                let unique_key = v.get_unique_key(&field.id);
                changes.push(UniqueChange {
                    a: 1,
                    key: unique_key.into(),
                    value: self.meta.id.0,
                });
            }
        }
        for value_rule in v_rules {
            match value_rule.detail.as_ref() {
                RecordRuleDetail::RrUnique(..) => {
                    let unique_key = self.get_unique_key_by_fids(value_rule);
                    changes.push(UniqueChange {
                        a: 1,
                        key: unique_key.into(),
                        value: self.meta.id.0,
                    });
                }
                _ => {}
            }
        }
    }
}

/// 计算一个模型现有的记录数
pub fn count_in_model(model_id: u64) -> usize {
    let key = VddKey::IndexModel2Record(model_id).to_string();
    let get_result = cache::get_index(&key);
    get_result.len()
}

/// 查询一个模型的记录中某个用户存放的记录数
fn count_in_model_by_uid(model_id: u64, operator: &UserId) -> usize {
    let mut r = 0usize;
    let index_key = VddKey::IndexModel2Record(model_id).to_string();
    for record_id in cache::get_index(&index_key).iter() {
        let meta = load_meta_from_rdb(&RecordId(*record_id)).unwrap();
        if &meta.uid == operator {
            r += 1;
        }
    }
    r
}

/// 加载一条记录元数据
pub fn load_meta_from_rdb(record_id: &RecordId) -> Option<RecordMeta> {
    let cf_meta = ROCKSDB.cf_handle(CF_META).unwrap();
    ROCKSDB
        .get_cf(
            &cf_meta,
            VddKey::EntityRecordMeta(record_id.clone()).to_string(),
        )
        .expect(i18n!(E_ROCKSDB).as_ref())
        .map(|v| serde_json::from_slice::<RecordMeta>(v.as_slice()).unwrap())
}

/// 存入一个上传文件的key和name的匹配关系
pub fn save_upload_file_name(operator: UserId, key: u64, file_name: &str) {
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    let vk = VddKey::EntityUploadFile(key);
    batch.put_cf(vk.db_name().unwrap(), vk.to_string(), file_name);
    log_db::save(
        &mut batch,
        &mut index_changes,
        &alog!(
            VddLogPart::LpAction(operator, "上传文件".into()),
            VddLogPart::LpFile(key, Arc::from(file_name))
        ),
    );
    ROCKSDB.write(batch).unwrap();
    cache::apply_index_change(&mut index_changes).unwrap();
}

/// 根据一个key读取一个上传文件的name
pub fn get_upload_file_name(operator: UserId, key: u64) -> String {
    let v = ROCKSDB
        .get(VddKey::EntityUploadFile(key).to_string())
        .expect(i18n!(E_ROCKSDB).as_ref());
    let file_name = String::from_utf8(v.unwrap()).unwrap();
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    log_db::save(
        &mut batch,
        &mut index_changes,
        &alog!(
            VddLogPart::LpAction(operator, "下载文件".into()),
            VddLogPart::LpFile(key, Arc::from(file_name.clone()))
        ),
    );
    ROCKSDB.write(batch).unwrap();
    cache::apply_index_change(&index_changes).unwrap();
    file_name
}

/// 加载一条模型数据记录
pub fn load_record(
    s: &VddSchema,
    record_id: &RecordId,
    pass_fids: &[FieldId],
) -> Option<VddRecord> {
    let meta = load_meta_from_rdb(record_id)?;
    let mut rv = HashMap::<FieldId, VddValueRef>::new();
    let cf = ROCKSDB.cf_handle(CF_FV).unwrap();
    s.fields.iter().for_each(|field| {
        if !pass_fids.contains(&field.id) {
            let fv = ROCKSDB
                .get_cf(
                    &cf,
                    VddKey::BoundFieldValue(field.id.clone(), meta.id.clone()).to_string(),
                )
                .expect(i18n!(E_ROCKSDB).as_ref())
                .map(|v| {
                    serde_json::from_slice::<VddValue>(v.as_slice()).unwrap_or(VddValue::NULL)
                });
            if fv.is_some() {
                rv.insert(field.id.clone(), fv.unwrap().into());
            }
        }
    });
    Some(VddRecord {
        meta: Arc::new(meta),
        value: rv,
    })
}

/// 加载一个模型的所有记录 用于导出模型
pub async fn select_records_all(model_id: u64) -> Vec<Arc<VddRecord>> {
    let schema = cache::get_schema(model_id).await.unwrap();
    let index_key = VddKey::IndexModel2Record(model_id).to_string();
    let mut cols = HashMap::<FieldId, Vec<VddValueWithMeta>>::new();
    schema.fields.iter().for_each(|field| {
        let fvs = value_db::fv_select_all(&field.id);
        if let Ok(fvs) = fvs {
            cols.insert(field.id.clone(), fvs);
        }
    });
    let mut record_vec = Vec::<Arc<VddRecord>>::new();
    if ROCKSDB.get(&index_key).unwrap().is_some() {
        for record_id in cache::get_index(&index_key).iter() {
            let meta = load_meta_from_rdb(&RecordId(*record_id)).unwrap();
            let mut hm = HashMap::<FieldId, VddValueRef>::new();
            schema.fields.iter().for_each(|field| {
                let vvm_null = VddValueWithMeta::new_null(RecordId(*record_id));
                let fv = cols
                    .get(&field.id)
                    .unwrap()
                    .iter()
                    .find(|vvm| vvm.rec_id.0 == *record_id)
                    .unwrap_or(&vvm_null);
                hm.insert(field.id.clone(), fv.value.clone());
            });
            record_vec.push(Arc::new(VddRecord {
                meta: Arc::new(meta),
                value: hm,
            }));
        }
    }
    record_vec.sort_by(|a, b| a.meta.id.0.cmp(&b.meta.id.0));
    record_vec
}

/// 清除一个模型内的所有记录
pub async fn clear_records(model_id: u64, operator_id: UserId) -> Ares {
    let schema = cache::get_schema(model_id).await?;
    let mut records = select_records_all(model_id).await;
    let v_rules = record_rule_db::rr_select_in_model(model_id).await;
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    let mut unique_changes = Vec::<UniqueChange>::new();
    for x in records.iter_mut() {
        do_delete(
            &mut batch,
            &mut index_changes,
            &mut unique_changes,
            &schema.model,
            Arc::get_mut(x).unwrap(),
            &v_rules,
            operator_id.clone(),
        )
        .await
    }
    let ares = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if ares.is_ok() {
        cache::apply_index_change(&index_changes)?;
        value_db::apply_unique_change(&unique_changes)?;
    }
    ares
}

/// 真正执行删除一笔记录的函数
async fn do_delete(
    batch: &mut WriteBatchWithTransaction<true>,
    index_changes: &mut Vec<IndexChange>,
    unique_changes: &mut Vec<UniqueChange>,
    model: &VddModel,
    record: &mut VddRecord,
    v_rules: &[Arc<RecordRule>],
    operator_id: UserId,
) {
    if model.uid != operator_id && record.meta.uid != operator_id {
        // 非模型作者与记录拥有着，需要验证是否授予权限
        let vg = visit_db::vg_try_load_in_model_by_user(model.id, &operator_id)
            .await
            .expect("没有访问权限");
        match record.fit_vg(&vg, &model.uid, &operator_id) {
            Ok(v) => {
                if !v.1 {
                    panic!("没有删除权限")
                }
            }
            Err(e) => {
                panic!("{}", e);
            }
        }
    }
    // 删除唯一性索引
    record.delete_unique_after_delete_record(&v_rules, unique_changes);
    // 删除当前记录值
    record.value.iter().for_each(|v| {
        let key = VddKey::BoundFieldValue(v.0.clone(), record.meta.id.clone());
        move_to_trash(batch, &key, None);
    });
    // 删除记录版本
    let kh = VddKey::BoundRecordAndVersion(record.meta.id.clone(), 0);
    delete_by_key_head(batch, &kh);
    // 删除记录基本信息
    let key = VddKey::EntityRecordMeta(record.meta.id.clone());
    move_to_trash(batch, &key, None);
    subscribe::send_change_event(batch, index_changes, model.id, &record.meta.id, AC_DELETE);
    index_changes.push(IndexChange::new(
        VddKey::IndexModel2Record(model.id),
        AC_DELETE,
        record.meta.id.0,
    ));
    log_db::save(
        batch,
        index_changes,
        &alog!(
            VddLogPart::LpAction(operator_id.to_owned(), "删除".into()),
            VddLogPart::LpRecord(record.meta.id.clone()),
        ),
    );
}

/// 删除一笔记录
pub async fn delete(record_id: &RecordId, schema: &VddSchema, operator_id: UserId) -> Ares {
    // 检验此用户是否能删除此记录
    let mut record = load_record(schema, record_id, &[]).unwrap();
    // 校验通过，可以删除
    let v_rules = record_rule_db::rr_select_in_model(schema.model.id).await;
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    let mut unique_changes = Vec::<UniqueChange>::new();
    do_delete(
        &mut batch,
        &mut index_changes,
        &mut unique_changes,
        &schema.model,
        &mut record,
        &v_rules,
        operator_id,
    )
    .await;
    let ares = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if ares.is_ok() {
        cache::apply_index_change(&index_changes)?;
        value_db::apply_unique_change(&unique_changes)?;
    }
    ares
}

/// 删除多笔记录
pub async fn delete_some(record_ids: &[RecordId], s: &VddSchema, operator_id: &UserId) -> Ares {
    let v_rules = record_rule_db::rr_select_in_model(s.model.id).await;
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    let mut unique_changes = Vec::<UniqueChange>::new();
    for record_id in record_ids {
        // 检验此用户是否能删除此记录
        let record = load_record(s, record_id, &[]);
        if let Some(mut record) = record {
            do_delete(
                &mut batch,
                &mut index_changes,
                &mut unique_changes,
                &s.model,
                &mut record,
                &v_rules,
                operator_id.clone(),
            )
            .await;
        }
    }
    let ares = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if ares.is_ok() {
        cache::apply_index_change(&index_changes)?;
        value_db::apply_unique_change(&unique_changes)?;
    }
    ares
}

/// 保存一条模型数据记录 用于新建记录，修改记录，回滚记录
pub async fn save(
    record: &mut VddRecord,
    schema: &VddSchema,
    v_rules: &[Arc<RecordRule>],
    old_to_version: bool,
) -> Ares {
    // tracing::info!("write record {} to rdb", self.meta.id.0);
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    let mut unique_changes = Vec::<UniqueChange>::new();
    let old_record = load_record(schema, &record.meta.id, &[]);
    let is_u = old_record.is_some();
    if is_u {
        let old_record = old_record.unwrap();
        old_record.delete_unique_after_delete_record(v_rules, &mut unique_changes);
        if old_to_version {
            record_version_db::create_version(&mut batch, &old_record).await;
        }
    }
    record.write_rdb(&schema, &mut batch, &mut index_changes);
    record.save_unique_key(&schema, &v_rules, &mut unique_changes);
    log_db::save(
        &mut batch,
        &mut index_changes,
        &alog!(
            VddLogPart::LpAction(
                record.meta.uid.to_owned(),
                if is_u {
                    "修改".into()
                } else {
                    "新建".into()
                }
            ),
            VddLogPart::LpModel(record.meta.model_id),
            VddLogPart::LpRecord(record.meta.id.clone()),
        ),
    );
    subscribe::send_change_event(
        &mut batch,
        &mut index_changes,
        schema.model.id,
        &record.meta.id,
        if is_u { AC_EDIT } else { AC_ADD },
    );
    let r = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if r.is_ok() {
        cache::apply_index_change(&index_changes)?;
        value_db::apply_unique_change(&unique_changes)?;
    }
    r
}

/// 非模型作者改动数据
pub async fn visit_save(
    record: &mut VddRecord,
    schema: &VddSchema,
    v_rules: &[Arc<RecordRule>],
    vg: &VisitGroup,
    is_u: bool,
) -> Ares {
    // tracing::info!("write record {} to rdb", self.meta.id.0);
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut changes = Vec::<FiledValueChange>::new();
    let mut index_changes = Vec::<IndexChange>::new();
    let mut unique_changes = Vec::<UniqueChange>::new();
    if is_u {
        let old_record = load_record(schema, &record.meta.id, &[]);
        if let Some(old_record) = old_record {
            old_record.delete_unique_after_delete_record(v_rules, &mut unique_changes);
            record_version_db::create_version(&mut batch, &old_record).await;
            old_record.value.iter().for_each(|x| {
                changes.push(FiledValueChange {
                    fid: x.0.clone(),
                    old_value: x.1.get_simple_display(),
                    new_value: "".into(),
                });
            });
        }
    } else {
        // 新建记录且配置了记录存放限制数量时，做数量校验
        let store_limit = vg.visit_store_limit();
        if store_limit > 0 {
            if count_in_model_by_uid(record.meta.model_id, &record.meta.uid) >= store_limit {
                return Err(format!("记录存放数量达到上限 {}", store_limit).into());
            }
        }
    }
    record.write_rdb(&schema, &mut batch, &mut index_changes);
    record.save_unique_key(&schema, &v_rules, &mut unique_changes);
    record.value.iter().for_each(|x| {
        changes.iter_mut().find(|y| y.fid == x.0.clone()).map(|y| {
            y.new_value = x.1.get_simple_display().into();
        });
    });
    changes.iter().for_each(|x| {
        if x.old_value != x.new_value {
            log_db::save(
                &mut batch,
                &mut index_changes,
                &alog!(
                    VddLogPart::LpAction(vg.u_id.clone(), "修改字段值".into()),
                    VddLogPart::LpModel(record.meta.model_id),
                    VddLogPart::LpField(x.fid.clone()),
                    VddLogPart::LpRecord(record.meta.id.clone()),
                    VddLogPart::LpFvChange(x.old_value.clone(), x.new_value.clone()),
                ),
            );
        }
    });
    log_db::save(
        &mut batch,
        &mut index_changes,
        &alog!(
            VddLogPart::LpAction(
                vg.u_id.clone(),
                if is_u {
                    "修改".into()
                } else {
                    "新增".into()
                }
            ),
            VddLogPart::LpModel(record.meta.model_id),
            VddLogPart::LpRecord(record.meta.id.clone()),
        ),
    );
    subscribe::send_change_event(
        &mut batch,
        &mut index_changes,
        schema.model.id,
        &record.meta.id,
        if is_u { AC_EDIT } else { AC_ADD },
    );
    let r = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if r.is_ok() {
        cache::apply_index_change(&index_changes)?;
    }
    r
}

/// 从parquet文件中导入一批记录
pub fn append_from_parquet(schema: &VddSchema, records: &[VddRecord]) -> Ares {
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    for record in records.iter() {
        record.write_rdb(schema, &mut batch, &mut index_changes);
    }
    let r = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if r.is_ok() {
        cache::apply_index_change(&index_changes)?
    }
    r
}

pub fn select_record_ids(ev: &VddValueEvaluation, rdb: &TransactionDB) -> Vec<RecordId> {
    let cf_fv = rdb.cf_handle(CF_FV).unwrap();
    let key_head = VddKey::BoundFieldValue(ev.get_field_id(), EMPTY_RID).to_string();
    let mut rec_id_s = Vec::<RecordId>::new();
    rdb.prefix_iterator_cf(&cf_fv, &key_head).for_each(|x| {
        let (k, v) = x.unwrap();
        let key = String::from_utf8(k.to_vec()).unwrap();
        if key.starts_with(&key_head) {
            let value = serde_json::from_slice::<VddValue>(&v).unwrap();
            if ev.evaluate(&value) {
                rec_id_s.push(RecordId(extract_u64_from_key(&key)));
            }
        }
    });
    rec_id_s
}

/// 通过一个列加载回收站中的记录
pub fn select_in_trash(fields: &[FieldId], _q: &WebQueryParams) -> Vec<VddRecordInTrash> {
    let cf_trash = ROCKSDB.cf_handle(CF_TRASH).unwrap();
    let mut iter = ROCKSDB.raw_iterator_cf(&cf_trash);
    let mut records = Vec::<VddRecordInTrash>::new();
    // 回收站中的字段值
    iter.seek_to_first();
    while iter.valid() {
        let (k, v) = iter.item().unwrap();
        let key = String::from_utf8(k.to_vec()).unwrap();
        let key1 = VddKey::try_from(key.as_str());
        if let Ok(key2) = key1 {
            match key2 {
                VddKey::BoundFieldValue(f, r) => {
                    if fields.contains(&f) {
                        let fv = serde_json::from_slice::<VddValue>(v).unwrap();
                        if records.iter().find(|rec| rec.record_id == r).is_none() {
                            records.push(VddRecordInTrash {
                                record_id: r.clone(),
                                value: HashMap::<FieldId, VddValueRef>::new(),
                            });
                        }
                        records
                            .iter_mut()
                            .find(|rec| rec.record_id == r)
                            .map(|rec| {
                                rec.value.insert(f.clone(), Arc::new(fv));
                            });
                    }
                }
                _ => {}
            }
        }
        iter.next();
    }
    records
}

pub fn drop_from_trash(record_ids: &[RecordId]) -> Ares {
    let cf_trash = ROCKSDB.cf_handle(CF_TRASH).unwrap();
    let mut batch = WriteBatchWithTransaction::<true>::default();
    for record_id in record_ids.iter() {
        let key = VddKey::EntityRecordMeta(record_id.clone()).to_string();
        batch.delete_cf(&cf_trash, key);
    }
    let mut iter = ROCKSDB.raw_iterator_cf(&cf_trash);
    iter.seek_to_first();
    while iter.valid() {
        let (k, _v) = iter.item().unwrap();
        let key = String::from_utf8(k.to_vec()).unwrap();
        let key = VddKey::try_from(key.as_str());
        if let Ok(key) = key {
            let key_s = key.to_string();
            match key {
                VddKey::BoundFieldValue(_f, r) => {
                    if record_ids.contains(&r) {
                        batch.delete_cf(&cf_trash, key_s);
                    }
                }
                _ => {}
            }
        }
        iter.next();
    }
    ROCKSDB.write(batch).map_err(|e| e.to_string().into())
}

fn restore_metas(model_id: u64, record_ids: &[RecordId]) -> Ares {
    let cf_trash = ROCKSDB.cf_handle(CF_TRASH).unwrap();
    let cf_meta = ROCKSDB.cf_handle(CF_META).unwrap();
    let mut batch = WriteBatchWithTransaction::<true>::default();
    let mut index_changes = Vec::<IndexChange>::new();
    for record_id in record_ids.iter() {
        let key = VddKey::EntityRecordMeta(record_id.clone()).to_string();
        match ROCKSDB.get_cf(&cf_trash, &key) {
            Ok(v) => match v {
                Some(v2) => {
                    batch.put_cf(&cf_meta, &key, v2);
                    batch.delete_cf(&cf_trash, &key);
                    index_changes.push(IndexChange::new(
                        VddKey::IndexModel2Record(model_id),
                        AC_ADD,
                        record_id.0,
                    ));
                }
                _ => {}
            },
            _ => {}
        }
    }
    let ares = ROCKSDB.write(batch).map_err(|e| e.to_string().into());
    if ares.is_ok() {
        cache::apply_index_change(&index_changes)?;
    }
    ares
}

/// 将一批记录从回收站恢复出去
pub async fn restore(model_id: u64, field_ids: &[FieldId], record_ids: &[RecordId]) -> Ares {
    if model_db::design_in_trash(model_id).await {
        return Err("需要模型和字段信息都不在回收站内才可以恢复数据".into());
    }
    // 基础信息
    restore_metas(model_id, record_ids)?;
    // 字段值
    value_db::fv_restore(field_ids, Some(record_ids))
}

mod display {
    use crate::value_db::VddValueDdTrait;
    use crate::{field_db, record_rule_db};
    use std::sync::Arc;
    use visit_dd_core::field::VddFieldType::FtComputed;
    use visit_dd_core::field::{ComputeAction, VddFieldType};
    use visit_dd_core::i18n;
    use visit_dd_core::model::VddSchema;
    use visit_dd_core::record::VddDisplay;
    use visit_dd_core::value::VddValue;

    fn find_max(field_type: &VddFieldType, values: &[Arc<VddValue>]) -> Arc<VddValue> {
        let mut idx = 0;
        let mut result_v: Arc<VddValue> = Arc::new(VddValue::S501 {});
        for value in values {
            if idx == 0 {
                result_v = value.to_owned();
            } else {
                let ordering = VddValue::compare(field_type, &value, &result_v);
                if ordering.is_ok() && ordering.unwrap().is_gt() {
                    result_v = value.to_owned();
                }
            }
            idx += 1;
        }
        result_v
    }

    fn find_min(field_type: &VddFieldType, values: &[Arc<VddValue>]) -> Arc<VddValue> {
        let mut idx = 0;
        let mut result_v: Arc<VddValue> = Arc::new(VddValue::S501 {});
        for value in values {
            if idx == 0 {
                result_v = value.to_owned();
            } else {
                let ordering = VddValue::compare(field_type, &value, &result_v);
                if ordering.is_ok() && ordering.unwrap().is_lt() {
                    result_v = value.to_owned();
                }
            }
            idx += 1;
        }
        result_v
    }

    /// 自动计算的值不存储到rocksdb中，每次查询时自动算
    async fn fill_display_501(schema: &VddSchema, display: &mut VddDisplay) {
        for field in schema.fields.iter() {
            if let FtComputed(action) = field.field_type.as_ref() {
                match action.as_ref() {
                    ComputeAction::CaConsumeCount => {
                        display.set_display(
                            &field.id,
                            &record_rule_db::rc_select(display.meta.id.clone(), 0, None)
                                .len()
                                .to_string(),
                        );
                    }
                    ComputeAction::CaConsumeCountByModel(consume_m_id) => {
                        display.set_display(
                            &field.id,
                            &record_rule_db::rc_select(
                                display.meta.id.clone(),
                                *consume_m_id,
                                None,
                            )
                            .len()
                            .to_string(),
                        );
                    }
                    ComputeAction::CaValueMaxInRecord(field_ids) => {
                        let fields = field_db::f_select_by_ids(field_ids).await;
                        let field_type = field_db::is_field_same_type(&fields);
                        if field_type.is_none() {
                            // 其实不需要，因为提交的配置都经过检验
                            display.set_display(&field.id, i18n!(1114).as_ref());
                        } else {
                            let field_type = field_type.unwrap();
                            let values = display.values_by_fields(&fields);
                            if values.is_empty() {
                                display.set_display(&field.id, i18n!(1114).as_ref());
                            } else {
                                let result_v = find_max(&field_type, &values);
                                display.set_display(&field.id, &result_v.get_simple_display());
                            }
                        }
                    }
                    ComputeAction::CaValueMinInRecord(field_ids) => {
                        let fields = field_db::f_select_by_ids(field_ids).await;
                        let field_type = field_db::is_field_same_type(&fields);
                        if field_type.is_none() {
                            display.set_display(&field.id, i18n!(1114).as_ref());
                        } else {
                            let field_type = field_type.unwrap();
                            let values = display.values_by_fields(&fields);
                            if values.is_empty() {
                                display.set_display(&field.id, i18n!(1114).as_ref());
                            } else {
                                let result_v = find_min(&field_type, &values);
                                display.set_display(&field.id, &result_v.get_simple_display());
                            }
                        }
                    }
                    ComputeAction::CaValueSumInRecord(field_ids) => {
                        let fields = field_db::f_select_by_ids(field_ids).await;
                        let field_type = field_db::is_field_same_type(&fields);
                        if field_type.is_none() {
                            // 其实不需要，因为提交的配置都经过检验
                            display.set_display(&field.id, i18n!(1114).as_ref());
                        } else {
                            let field_type = field_type.unwrap();
                            let values = display.values_by_fields(&fields);
                            match field_type.as_ref() {
                                VddFieldType::FtInt { .. } => {
                                    let mut v = 0i64;
                                    values.iter().for_each(|x| {
                                        v += x.as_ref().try_into().unwrap_or(0i64);
                                    });
                                    display.set_display(&field.id, &v.to_string());
                                }
                                VddFieldType::FtFloat { .. } => {
                                    let mut v = 0f64;
                                    values.iter().for_each(|x| {
                                        v += x.as_ref().try_into().unwrap_or(0f64);
                                    });
                                    display.set_display(&field.id, &format!("{:.1$}", v, 4));
                                }
                                _ => {
                                    display.set_display(&field.id, "发现不适合参与计算的字段");
                                }
                            }
                        }
                    }
                    ComputeAction::CaValueAvgInRecord(field_ids) => {
                        let fields = field_db::f_select_by_ids(field_ids).await;
                        let field_type = field_db::is_field_same_type(&fields);
                        if field_type.is_none() {
                            // 其实不需要，因为提交的配置都经过检验
                            display.set_display(&field.id, i18n!(1114).as_ref());
                        } else {
                            let field_type = field_type.unwrap();
                            let values = display.values_by_fields(&fields);
                            match field_type.as_ref() {
                                VddFieldType::FtInt { .. } | VddFieldType::FtFloat { .. } => {
                                    let mut v = 0f64;
                                    values.iter().for_each(|x| {
                                        v += x.as_ref().try_into().unwrap_or(0f64);
                                    });
                                    v = v / fields.len() as f64;
                                    display.set_display(&field.id, &format!("{:.1$}", v, 4));
                                }
                                _ => {
                                    display.set_display(&field.id, "发现不适合参与计算的字段");
                                }
                            }
                        }
                    }
                }
            }
        }
    }

    pub(super) async fn fill_display(
        display: &mut VddDisplay,
        schema: &VddSchema,
        fill_meta: bool,
    ) {
        if fill_meta {
            display.meta_to_display();
        }
        for (fid, fv) in display.value.iter() {
            display
                .display
                .insert(fid.clone(), fv.to_display(fid, &schema).await);
        }
        fill_display_501(&schema, display).await;
    }
}
